package com.key.w8.http1.connection.connectionImpl;

import com.key.w8.common.execute.ChainExecuteHandler;
import com.key.w8.http1.entity.HttpRequest;
import com.key.w8.http1.entity.HttpResponse;
import com.key.w8.http1.exception.HttpResponseTimeOutException;
import com.key.w8.http1.exception.RequestException;
import com.key.w8.http1.exception.ResponseException;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.key.w8.http1.constant.Constants.ACK_HEADER;

/**
 * 改双向为单向，仅支持客户端向服务器
 *
 * H2支持多路复用，故此所有请求的提交都是非阻塞的
 * H1长连接故为了保证客户端的一致性，客户端都使用FutureTask提交写任务，连接建立后，不断从队列里拿任务
 * 请求队列不使用阻塞队列是因为，请求会大量发送，如果使用阻塞队列会涉及大量的锁操作,使用cas替代
 * 而响应由于是一个请求对应一个响应，一次处理只会poll一次，所以相对开销较小可以使用
 *
 * @Author: k8
 * @CreateTime: 2025-02-28
 * @Version: 1.0
 */
@Slf4j
public class ClientHttp1PersistentConnection extends DefaultHttp1Connection {

    private ConcurrentLinkedDeque<FutureTask<HttpResponse>> tasks;
    private AtomicInteger requestCounter;
    private Lock postClosedLock = new ReentrantLock();
    protected Lock reconnectLock = new ReentrantLock();

    /**
     * http1协议如果需要响应，则会尝试从post中获取，并有等待响应阻塞的超时时间responseTimeOut
     * 当接收到响应后，应该放到post中，而futureTask获取到之后便相当于本次请求完整结束
     */
    private BlockingQueue<HttpResponse> responsePost = new LinkedBlockingQueue<>();//驿站
    /**
     * 不同于Tcp协议的ack响应，这里的responseTimeOut是指
     */
    private final long responseTimeOut;
    private final TimeUnit responseTimeUnit;
    private AtomicLong ackCounter = new AtomicLong(0);


    /**
     * 取消使用阻塞队列的形式，当大量的请求打来时
     */
    private Semaphore requestSemaphore = new Semaphore(0);
    //是否最大发送消息是有界的
    private final boolean bounded;
    private int maxRequestSends;
    private volatile boolean postClosed = false;

    public ClientHttp1PersistentConnection(long responseTimeOut, TimeUnit responseTimeUnit,int maxRequestSends) {
        check(responseTimeOut,responseTimeUnit,maxRequestSends);
        this.responseTimeOut = TimeUnit.MILLISECONDS.convert(responseTimeOut,responseTimeUnit);
        this.responseTimeUnit = TimeUnit.MILLISECONDS;
        tasks = new ConcurrentLinkedDeque<>();
        this.requestCounter = new AtomicInteger(0);
        this.maxRequestSends = maxRequestSends;
        this.bounded = true;

        super.addChainExecuteHandler(new ChainExecuteHandler(){
            @Override
            public Object onInvoke(Object msg, Object attach) {
                if (msg instanceof HttpResponse){
                    HttpResponse response = (HttpResponse) msg;
                    try {
                        responsePost.put(response);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return super.fireNextHandler(response,attach);
                }else {
                    throw new IllegalArgumentException("msg must is FullHttpResponse");
                }
            }
        });
    }

    public ClientHttp1PersistentConnection(long responseTimeOut, TimeUnit responseTimeUnit) {
        assert responseTimeOut > 0;
        assert responseTimeUnit != null;
        this.responseTimeOut = TimeUnit.MILLISECONDS.convert(responseTimeOut,responseTimeUnit);
        this.responseTimeUnit = TimeUnit.MILLISECONDS;
        tasks = new ConcurrentLinkedDeque<>();
        this.requestCounter = new AtomicInteger(0);
        this.bounded = false;
        super.addChainExecuteHandler(new ChainExecuteHandler(){
            @Override
            public Object onInvoke(Object msg, Object attach) {
                if (msg instanceof HttpResponse){
                    HttpResponse response = (HttpResponse) msg;
                    try {
                        responsePost.put(response);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    return super.fireNextHandler(response,attach);
                }else {
                    throw new IllegalArgumentException("msg must is FullHttpResponse");
                }
            }
        });
    }


    private void check(long responseTimeOut, TimeUnit responseTimeUnit,int maxRequestSends) {
        assert responseTimeOut > 0;
        assert responseTimeUnit!=null;
        assert maxRequestSends > 0;
    }
    public HttpResponse writeRequestNeedResponseSync(HttpRequest httpRequest) throws ExecutionException, InterruptedException {

        ChannelPromise promise = getPromise();
        super.writeRequest(httpRequest,promise);
        promise.get();
        if (promise.isSuccess()){
            long responseTimeOut = this.responseTimeOut;
            return responsePost.poll(responseTimeOut, responseTimeUnit);
        }else {
            throw new RequestException("Send request failed.");
        }
    }

    /**
     * 支持异步发送请求，将请求包装成futureTask,提交到任务池，返回该futureTask，观测执行情况
     * 这也是第一次在非线程池环境中使用FutureTask
     * 这也是和CompletableFuture最大的区别，CompletableFuture虽然支持异步的执行任务，但是它本身并不代表一个任务
     * 而FutureTask本身是一个支持返回消息的任务，两者不应该是互斥的关系，应该是合作使用
     * @Author: k8
     * @date:
     * @param: httpRequest
     * @return: java.util.concurrent.FutureTask<io.netty.handler.codec.http.HttpResponse>
     */
    public Future<HttpResponse> writeRequestNeedResponse(HttpRequest httpRequest){
        FutureTask<HttpResponse> futureTask = new FutureTask<>(()->{
            long startTime = System.currentTimeMillis();
            ChannelPromise promise = getPromise();
            long count = startTime+ackCounter.incrementAndGet();
            httpRequest.headers().add(ACK_HEADER,count);
            super.writeRequest(httpRequest,promise);
            try {
                promise.get();
            }catch (Throwable throwable){
                throw throwable;
            }
            if (promise.isSuccess()){
                HttpResponse poll = null;
                long responseTimeOut = this.responseTimeOut;
                //丢弃ack不对的poll
                while (responseTimeOut > 0){
                    if (state().value >= State.CLOSED.value){
                        throw new ResponseException("Response get failed, request uri:"+httpRequest.uri()+", start Time: "+startTime);
                    }
                    try{
                        poll = responsePost.poll(responseTimeOut, responseTimeUnit);
                    }catch (InterruptedException e){
                        throw new ResponseException("Response get failed, request uri:"+httpRequest.uri()+", start Time: "+startTime);
                    }
                    if (poll != null){
                        HttpHeaders headers = poll.headers();
                        if (headers.contains(ACK_HEADER) && Long.parseLong(headers.get(ACK_HEADER)) == count){
                            checkResponse(poll);
                            return poll;
                        }else {
                            responseTimeOut = responseTimeOut-(System.currentTimeMillis()-startTime);
                        }
                    }else {
                        throw new ResponseException("Response get failed, request uri:"+httpRequest.uri()+", start Time: "+startTime);
                    }
                }
                throw new HttpResponseTimeOutException("StartTime:"+startTime);
            }else {
                throw new RequestException("Send request failed.");
            }
        });
        //此时还没有添加，如果不是created状态则直接取消
        if (state().value <= State.CREATED.value){//第一次检查状态
            if (bounded && requestCounter.get() < maxRequestSends){
                int i = requestCounter.incrementAndGet();
                if (i > maxRequestSends){
                    futureTask.cancel(false);
                    requestCounter.decrementAndGet();
                }else {
                    if (state().value <= State.CREATED.value){//recheck
                        futureTask.cancel(false);
                        requestCounter.decrementAndGet();
                        return futureTask;
                    }
                    this.tasks.offer(futureTask);
                    requestSemaphore.release();
                }
            }else if (!bounded){
                requestCounter.incrementAndGet();
                //recheck是必须的，确保在占用资源期间没有shutDown或close
                //如果是shutDown，不好判断是否已经执行完了处理流程
                //如果不recheck
                //当在占用成功之前执行完的，这里添加的任务将不会得到处理，就会出现问题，所以需要recheck
                //如果是占用成功之后执行完的，这里添加就没问题，但没必要再添加了，在Java线程池里如果是队头有元素就添加了
                if (state().value > State.CREATED.value){//recheck
                    futureTask.cancel(false);
                    requestCounter.decrementAndGet();
                    return futureTask;
                }
                this.tasks.offer(futureTask);
                requestSemaphore.release();
                if (state().value > State.SHUTDOWN.value){
                    try {
                        close();
                    } catch (IOException e) {
                    }
                    futureTask.cancel(false);
                }
            }else {
                futureTask.cancel(false);
            }
        }else {
            futureTask.cancel(false);
        }
        return futureTask;
    }

    private void checkResponse(HttpResponse response) {
        if (response.status().code() > 299 || response.status().code()<200){
            if (StringUtil.isNullOrEmpty(response.message())){
                throw new ResponseException("Response status error, status:"+response.status().code());
            }else {
                throw new ResponseException(response.message());
            }
        }
    }

    /**
     * 取消tasks里的所有任务
     * 但可能被多个线程包括守护线程同时调用，所以为了避免一直竞争阻塞队列里的锁
     * 需要有一把取消锁，将取消权限交给抢到锁的线程
     *
     * 如果是一个线程提交过后发现状态不对，进行取消的，那么它的任务必定已经提交进来
     * 这样做可以避免每次提交任务都需要去上锁
     */
    private void cancelTasks(boolean mayInterruptIfRunning){
        while (requestCounter.get()>0 && !tasks.isEmpty()){
            FutureTask<HttpResponse> poll = tasks.poll();
            if (poll != null){
                requestCounter.decrementAndGet();
                if (!poll.isDone()){
                    poll.cancel(mayInterruptIfRunning);
                }
            }
        }
        while (requestCounter.get()>0 || !tasks.isEmpty()){
            FutureTask<HttpResponse> poll = tasks.poll();
            if (poll != null){
                requestCounter.decrementAndGet();
                if (!poll.isDone()){
                    poll.cancel(mayInterruptIfRunning);
                }
            }
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 使用FutureTask的一个好处是，当出现什么读写异常了，可以保留异常，并不上抛
     */
    public void runOut(){
        for (;;){
            //pool是先中断再关闭，这样的原因是关闭线程池时可能某个任务一直在等响应，而close必须保证任务被取消完，包括在执行任务，此时如果
            //不先中断让其提前结束，则close一直无法执行完，中断也一直没办法执行，虽然在得到响应后可以正常中断，但是不够快
            if ((state().value > State.SHUTDOWN.value) || (state() == State.SHUTDOWN && requestCounter.get() == 0) || Thread.currentThread().isInterrupted()) break;
            FutureTask<HttpResponse> task = null;
            try {
                requestSemaphore.acquire();
            } catch (InterruptedException e) {
                break;
            }
            try {
                try {
                    reconnectLock.lock();
                }finally {
                    reconnectLock.unlock();
                }
                task = tasks.poll();
                if (task != null){
                    if (state().value > State.SHUTDOWN.value){//尽量减少超时等待
                        task.cancel(false);
                        break;
                    }else {
                        task.run();
                    }
                }
            }finally {
                if (task != null) {
                    requestCounter.decrementAndGet();
                }
            }
        }
    }

    @Override
    public void close() throws IOException {
        super.close();
        try {
            postClosedLock.lock();
            if (!postClosed){
                cancelTasks(false);
            }
        }finally {
            postClosedLock.unlock();
        }
    }

    @Override
    public void shutDown() throws IOException {
        if (state().value < State.SHUTDOWN.value){
            state(State.SHUTDOWN);
        }
    }

}
